using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Protocol;
using MQTTnet.Packets;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;
using System.Diagnostics;

public class MqttClientWrapper : IDisposable
{
    private readonly IMqttClient _mqttClient;
    // 修改_tagStore数据结构，添加时间戳
    private readonly ConcurrentDictionary<string, (string value, DateTime lastUpdate)> _tagStore = 
        new ConcurrentDictionary<string, (string, DateTime)>();
    private readonly Channel<MqttApplicationMessage> _messageChannel;
    private readonly CancellationTokenSource _cancellationTokenSource;
    private bool _isDisposed;
    private string _lastBrokerAddress = string.Empty;
    private int _lastPort = 1883;
    private bool _isDisposing;
    private readonly List<string> _activeSubscriptions;
    private long _processedCount;
    private Stopwatch _stopwatch = Stopwatch.StartNew();
    // 定义事件
    public event Action Connected;
    public event Action<string> Disconnected;
    public event Action<string> SubscriptionAdded;
    public event Action<string> SubscriptionRemoved;

    public MqttClientWrapper()
    {
        var factory = new MqttFactory();
        _mqttClient = factory.CreateMqttClient();
        // 更新为新的类型定义
        _tagStore = new ConcurrentDictionary<string, (string value, DateTime lastUpdate)>();
        // 修改通道创建方式（在构造函数中）
        _messageChannel = Channel.CreateBounded<MqttApplicationMessage>(new BoundedChannelOptions(100000)
        {
            FullMode = BoundedChannelFullMode.DropOldest, // 超限时丢弃旧消息
            SingleWriter = true,
            SingleReader = false
        });
        _cancellationTokenSource = new CancellationTokenSource();
        _activeSubscriptions = new List<string>();
    SetupEventHandlers();
    LogMessage("MqttClientWrapper initialized.");
    StartMessageProcessing();
}
    private void SetupEventHandlers()
    {
        _mqttClient.ConnectedAsync += async args => {
            LogMessage("Connected to the broker.");
            Connected?.Invoke();
            await RestoreSubscriptionsAsync();
        };

        _mqttClient.DisconnectedAsync += async args => {
            string reason = args.Reason.ToString();
            LogMessage($"Disconnected from the broker: {reason}");
            Disconnected?.Invoke(reason);
            if (!_isDisposing) {
                await Task.Delay(5000);
                await SafeReconnectAsync();
            }
        };

        _mqttClient.ApplicationMessageReceivedAsync += e => {
            _messageChannel.Writer.TryWrite(e.ApplicationMessage);
            return Task.CompletedTask;
        };
    }

    private void StartMessageProcessing()
    {
        const int MaxDegreeOfParallelism = 8; // 根据CPU核心数调整
        var options = new ParallelOptions { MaxDegreeOfParallelism = MaxDegreeOfParallelism };
        
        Task.Run(async () => {
            while (await _messageChannel.Reader.WaitToReadAsync(_cancellationTokenSource.Token))
            {
                var messages = new List<MqttApplicationMessage>();
                
                // 批量读取消息
                while (_messageChannel.Reader.TryRead(out var message))
                {
                    messages.Add(message);
                    if (messages.Count >= 10000) break; // 每批最多处理1000条
                }
        
                // 并行处理
                Parallel.ForEach(messages, options, msg => 
                {
                    try 
                    {
                        ProcessMessage(msg);
                    }
                    catch (Exception ex)
                    {
                        LogMessage($"Error processing message: {ex.Message}");
                    }
                });
            }
        }, _cancellationTokenSource.Token);
    }
    private void ProcessMessage(MqttApplicationMessage message)
    {
        string topic = message.Topic;
        string payload = Encoding.UTF8.GetString(message.Payload);
        _tagStore[topic] = (payload, DateTime.UtcNow);
        //LogMessage($"Updated tag [{topic}] at {DateTime.UtcNow:HH:mm:ss.fff}"); // 添加精确时间日志
        //OnTagUpdated(topic, payload);

        //Interlocked.Increment(ref _processedCount);
        //// 每5秒输出一次吞吐量
        //if (_stopwatch.Elapsed.TotalSeconds >= 5)
        //{
        //    var count = Interlocked.Exchange(ref _processedCount, 0);
        //    LogMessage($"Processing rate: {count / _stopwatch.Elapsed.TotalSeconds:F2} msg/s");
        //    _stopwatch.Restart();
        //}
    }
    public IDictionary<string, string> GetAllTags(TimeSpan? maxAge = null)
    {
        var now = DateTime.UtcNow;
        var maxAgeValue = maxAge ?? TimeSpan.FromSeconds(10); // 默认5秒内数据为有效
        
        return _tagStore.ToDictionary(kvp => kvp.Key, kvp => kvp.Value.value);
    }
    private void OnTagUpdated(string tag, string value)
    {
        LogMessage($"Tag {tag} updated with value: {value}");
    }

    // 客户端设置 Clean Session = false 时，服务器保存订阅状态和未确认的QoS 1/2消息，断线重连后恢复，适用于移动设备网络不稳定的场景
    public async Task ConnectAsync(string broker = "localhost", int port = 1883, string username = "admin", string password = "password", bool cleanSession = true)
    {
        if (_mqttClient.IsConnected) {
            LogMessage("Already connected");
            return;
        }

        try {
            _lastBrokerAddress = broker;
            _lastPort = port;
            var optionsBuilder = new MqttClientOptionsBuilder()
                .WithTcpServer(broker, port)
                .WithClientId($"CLIENT_{Guid.NewGuid().ToString("N").Substring(0, 8)}")
                .WithCleanSession(cleanSession)
                .WithKeepAlivePeriod(TimeSpan.FromSeconds(30)); // 客户端定期发送心跳包维持连接，超时未收到则服务器主动断开，避免“僵尸连接”

            if (!string.IsNullOrEmpty(username)) {
                optionsBuilder.WithCredentials(username, password);
            }

            var options = optionsBuilder.Build();
            await _mqttClient.ConnectAsync(options);
            LogMessage("Connection established.");
        }
        catch (Exception ex) {
            LogMessage($"Connection failed: {ex.Message}. Retrying...");
            await SafeReconnectAsync();
        }
    }

    public async Task DisconnectAsync(bool graceful = true)
    {
        if (!_mqttClient.IsConnected) {
            LogMessage("Already disconnected");
            return;
        }
        var options = graceful ? new MqttClientDisconnectOptions() : new MqttClientDisconnectOptions();
        await _mqttClient.DisconnectAsync(options);
        LogMessage("Disconnected gracefully.");
    }

    private async Task SafeReconnectAsync()
    {
        int retryCount = 0;
        int maxRetries = 5;
        int baseDelay = 1000;
        int maxDelay = 30000;
        while (retryCount < maxRetries) {
            try {
                if (!_mqttClient.IsConnected) {
                    await ConnectAsync(_lastBrokerAddress, _lastPort, "admin", "password");
                    break;
                }
            }
            catch (Exception ex) {
                retryCount++;
                int delay = Math.Min(baseDelay * (int)Math.Pow(2, retryCount), maxDelay);
                LogMessage($"Reconnect failed (attempt {retryCount}): {ex.Message}. Retrying in {delay} ms...");
                await Task.Delay(delay);
            }
        }
        if (!_mqttClient.IsConnected) {
            LogMessage("Failed to reconnect after multiple attempts.");
        }
    }

    private async Task RestoreSubscriptionsAsync()
    {
        var existingSubs = new List<string>(); // 实际中需要从持久化存储或其他地方获取
        if (existingSubs.Count > 0) {
            await SubscribeBulkAsync(existingSubs);
        }
    }

    // AtMostOnce = 0x00, AtLeastOnce = 0x01, ExactlyOnce = 0x02
    // QoS级别：三种服务质量，确保消息可靠传输
    // 保留消息（Retained Messages）：服务器保存最后一条消息供新订阅者获取
    public async Task PublishAsync(string topic, string payload, MqttQualityOfServiceLevel qos = MqttQualityOfServiceLevel.AtMostOnce, bool retain = true)
    {
        EnsureConnected();
        var message = new MqttApplicationMessageBuilder()
            .WithTopic(topic)
            .WithPayload(payload)
            .WithQualityOfServiceLevel(qos)
            .WithRetainFlag(retain)
            .Build();

        await _mqttClient.PublishAsync(message);
    }
    public async Task PublishBulkAsync(IEnumerable<(string topic, string payload, MqttQualityOfServiceLevel qos, bool retain)> messages)
    {
        EnsureConnected();

        var tasks = messages.Select(message => 
            _mqttClient.PublishAsync(new MqttApplicationMessageBuilder()
                .WithTopic(message.topic)
                .WithPayload(message.payload)
                .WithQualityOfServiceLevel(message.qos)
                .WithRetainFlag(message.retain)
                .Build()));
        
        await Task.WhenAll(tasks);
    }


    public async Task PublishBatchParallelAsync(IEnumerable<KeyValuePair<string, string>> topics, int maxConcurrency = 10)
    {
        var semaphore = new SemaphoreSlim(maxConcurrency);
        var tasks = topics.Select(async topic =>
        {
            await semaphore.WaitAsync();
            try
            {
                await PublishAsync(topic.Key, topic.Value);
            }
            finally
            {
                semaphore.Release();
            }
        });

        await Task.WhenAll(tasks);
    }

    private MqttTopicFilter BuildMqttTopicFilter(string topic, MqttQualityOfServiceLevel qos)
    {
        return new MqttTopicFilterBuilder()
            .WithTopic(topic)
            .WithQualityOfServiceLevel(qos)
            .Build();
    }

    private async Task SubscribeTopicsAsync(IEnumerable<string> topics, MqttQualityOfServiceLevel qos)
    {
        var subscribeOptionsBuilder = new MqttClientSubscribeOptionsBuilder();
        foreach (var topic in topics) {
            var filter = BuildMqttTopicFilter(topic, qos);
            subscribeOptionsBuilder = subscribeOptionsBuilder.WithTopicFilter(filter);
        }
        var subscribeOptions = subscribeOptionsBuilder.Build();
        try {
            await _mqttClient.SubscribeAsync(subscribeOptions);
        }
        catch (Exception ex) {
            LogMessage($"Error while subscribing to topics: {string.Join(", ", topics)}. Error: {ex.Message}");
        }
    }

    public async Task SubscribeBulkAsync(IEnumerable<string> topics, MqttQualityOfServiceLevel qos = MqttQualityOfServiceLevel.AtLeastOnce)
    {
        IEnumerable<string> uniqueTopics = topics
            .Distinct()
            .Where(t => !_activeSubscriptions.Contains(t));
        if (!uniqueTopics.Any()) return;
        await SubscribeTopicsAsync(uniqueTopics, qos);
        foreach (var topic in uniqueTopics) {
            _activeSubscriptions.Add(topic);
            SubscriptionAdded?.Invoke(topic);
            LogMessage($"Successfully subscribed to topic: {topic}");
        }
    }

    public async Task UnsubscribeBulkAsync(params string[] topics)
    {
        var unsubscribeOptionsBuilder = new MqttClientUnsubscribeOptionsBuilder();
        foreach (var topic in topics) {
            unsubscribeOptionsBuilder = unsubscribeOptionsBuilder.WithTopicFilter(topic);
        }
        var unsubscribeOptions = unsubscribeOptionsBuilder.Build();
        try {
            await _mqttClient.UnsubscribeAsync(unsubscribeOptions);
        }
        catch (Exception ex) {
            LogMessage($"Error while unsubscribing from topics: {string.Join(", ", topics)}. Error: {ex.Message}");
        }
        foreach (var topic in topics) {
            _activeSubscriptions.Remove(topic);
            SubscriptionRemoved?.Invoke(topic);
            LogMessage($"Successfully unsubscribed from topic: {topic}");
        }
    }

    public IEnumerable<string> GetActiveSubscriptions()
    {
        return _activeSubscriptions.AsReadOnly();
    }

    private void LogMessage(string message)
    {
        Debug.WriteLine($"[MqttClientWrapper] {message}");
    }

    private void EnsureConnected()
    {
        if (!_mqttClient.IsConnected) {
            throw new InvalidOperationException("MQTT client is not connected.");
        }
    }

    public void Dispose()
    {
        Dispose(true);
        GC.SuppressFinalize(this);
    }

    protected virtual void Dispose(bool disposing)
    {
        if (!_isDisposed) {
            if (disposing) {
                _isDisposing = true;
                DisconnectAsync().ConfigureAwait(false).GetAwaiter().GetResult();
                _cancellationTokenSource.Cancel();
                _cancellationTokenSource.Dispose();
                _mqttClient.Dispose();
            }
            _isDisposed = true;
        }
    }

    public bool TryGetTagValue(string tag, out string value)
    {
        if (_tagStore.TryGetValue(tag, out var result))
        {
            value = result.value;
            return true;
        }
        value = default;
        return false;
    }
    public bool IsConnected => _mqttClient.IsConnected;
}